package associationrules;

import static util.DataUtil.incrementArray;
import static util.Preconditions.check;
import static util.Preconditions.checkNotNull;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import util.CharacterSet;
import util.Pair;

/**
 * 
 * Solves the association rules problem using the MapReduce framework
 *
 */
public class AssociationRulesMapReduce {
	
	private long minOccurrences;
	private int size;
	private int sliceSize;
	
	public AssociationRulesMapReduce(long minOccurrences, int size, int sliceSize) {
		check(minOccurrences > 0, "minOccurrences must be positive");
		check(size > 0, "size must be positive");
		check(sliceSize > 0, "sliceSize must be positive");
		
		this.minOccurrences = minOccurrences;
		this.size = size;
		this.sliceSize = sliceSize;
	}

	/**
	 * Splits the input to be processed by each mapper
	 */
	public List<String> split(String source) {
		// FIXME add checks
		List<String> slices = new LinkedList<String>();
		source = source.replaceAll(" ", "");
		System.out.println(source);
		int start = 0;
		int end = 0;
		
		for (start = 0, end = sliceSize - 1; end < source.length(); ) {
			slices.add(source.substring(start, end + 1));
			// if reaches the end ...
			if (end == source.length() - 1) {
				// force the for condition to break
				end++;
			} else {
				start = end - (size - 1) + 1;
				end = Math.min(end + sliceSize, source.length() - 1);
			}
		}
		
		return slices;
	}
	
	/**
	 * From MapReduce framework
	 * 
	 * input : List<Pair<K1, V1>>
	 * output : List<Pair<K2, V2>>
	 * 
	 * We don't need the inputs keys, so the input is only a String.
	 */
	public List<Pair<CharacterSet, Long>> map(String source) {
		checkNotNull(source, "source must not be null.");
		// removes the whitespace between the characters
		source = source.substring(0, source.length()).replaceAll(" ", "");
		List<Pair<CharacterSet, Long>> rules = new ArrayList<Pair<CharacterSet, Long>>();
		int[] indexes = new int[size];
		
		// initialises the index array
		for (int i = 0; i < indexes.length; i++) {
			indexes[i] = i;
		}
		
		// add the first combination
		CharacterSet characters = new CharacterSet();
		for (int index : indexes) {
			characters.add(source.charAt(index));
		}
		rules.add(new Pair<CharacterSet, Long>(characters, 1L));
		
		// add the remaining combinations
		while (incrementArray(indexes, 0, source.length() - size)) {
			characters = new CharacterSet();
			for (int index : indexes) {
				characters.add(source.charAt(index));
			}
			rules.add(new Pair<CharacterSet, Long>(characters, 1L));
		}
		
		return rules;
	}
	
	/**
	 * Prepares the maps's outputs to be processed by the reducers 
	 */
	public Map<CharacterSet, List<Long>> join(List<Pair<CharacterSet, Long>> input) {
		Map<CharacterSet, List<Long>> result = new HashMap<CharacterSet, List<Long>>();
		
		for (Pair<CharacterSet, Long> pair : input) {
			if (result.get(pair.getFirst()) == null) {
				result.put(pair.getFirst(), new LinkedList<Long>());
			}
			
			result.get(pair.getFirst()).add(pair.getSecond());
		}
		
		return result;
	}
	
	/**
	 * From MapReduce framework
	 * 
	 * input : Map<K2, List<V2>>
	 * output : Map<K3, V3>
	 */
	public Map<CharacterSet, Long> reduce(Map<CharacterSet, List<Long>> input) {
		checkNotNull(input, "input must not be null.");
		Map<CharacterSet, Long> aux = new HashMap<CharacterSet, Long>();
		Map<CharacterSet, Long> result = new HashMap<CharacterSet, Long>();
		
		for (CharacterSet key : input.keySet()) {
			Long newKey = 0L;
			
			for (Long occurrence : input.get(key)) {
				newKey += occurrence;
			}
			
			aux.put(key, newKey);
		}
		
		for (CharacterSet key : aux.keySet()) {
			if (aux.get(key) >= minOccurrences) {
				result.put(key, aux.get(key));
			}
		}
		
		return result;
	}
}
